package com.yifeng.repo.flink.data.transport.bootstrap;

import com.yifeng.repo.flink.data.transport.config.*;
import com.yifeng.repo.flink.data.transport.streaming.iceberg.FlinkSinkIcebergBuilder;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 通过DataStream写入数据
 * @author wangzhi
 * @since 2023-06-07
 */
public class FlinkWriteToIceberg {

    public static void main(String[] args) throws Exception {
        //获取配置
        JobParametersConfig config = BootstrapSourceKafkaHelper.setEnvAndParameter(args);
        StreamExecutionConfig streamExecutionConfig = config.getStreamExecutionConfig();
        KafkaConfig kafkaConfig = config.getKafkaConfig();
        SinkIcebergConfig sinkIcebergConfig = config.getSinkIcebergConfig();
        //1.必须设置checkpoint ,Flink向Iceberg中写入数据时当checkpoint发生后，才会commit数据。
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(streamExecutionConfig.getCheckpointInterval());

        //2.读取Kafka 中的topic 数据
        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers(kafkaConfig.getBootstrapServers())
                .setTopics(kafkaConfig.getTopic())
                .setGroupId(kafkaConfig.getGroupId())
                .setStartingOffsets(OffsetsInitializer.latest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();
        DataStreamSource<String> dataStreamSource = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");


        //3.构建写入Iceberg表处理逻辑定义。
        FlinkSinkIcebergBuilder.buildFlinkSinkIceberg(dataStreamSource, sinkIcebergConfig);


        //4.启动任务
        env.execute(config.getCdcTaskname());
    }



}
